package day04

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 2.4.4	案例实操（广告点击Top3）
 *
 * 统计出每一个省份广告被点击次数的Top3
 *
 * 输入：
 * 时间戳             省份id      城市id        用户id       广告id
 * 1516609143867      6          7             64          16
 * 1516609143869      9          4             75          18
 * 1516609143869      1          7             87          12
 * 1516609143869      2          8             92          9
 *
 */
object TransformationPractice {

  def main(args: Array[String]): Unit = {

    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    //3.创建RDD
    val agentFileRdd: RDD[String] = sc.textFile("input/agent.log")

    // 查看原始数据格式
    // printContent(agentFileRdd)

    // 4.转换原始数据的格式
    val newRdd: RDD[(String, Int)] = agentFileRdd.map(str => {
      val strings: Array[String] = str.split(" ")
      // 省份和广告拼接的字符串
      val provinceAndAdvertisement: String = strings(1).concat("-").concat(strings(4))
      (provinceAndAdvertisement, 1)
    })

    // 5. 获取省份+广告 被点击次数
    // 江苏省-A广告 3次
    // 江苏省-B广告 5次
    // 安徽省-A广告 8次
    // 安徽省-C广告 2次
    val proAdvertiseRdd: RDD[(String, Int)] = newRdd.reduceByKey(_ + _)

    //    proAdvertiseRdd.collect().foreach(println)

    val provinceRdd: RDD[(String, (String, Int))] = proAdvertiseRdd.map {
      case (str, i) =>
        val strings: Array[String] = str.split("-")
        (strings(0), (strings(1), i))
    }

    val provinceGroupRdd: RDD[(String, Iterable[(String, Int)])] = provinceRdd.groupByKey()

    val resultRdd: RDD[(String, List[(String, Int)])] = provinceGroupRdd.mapValues(iterable => {
      val list: List[(String, Int)] = iterable.toList
      val tuples: List[(String, Int)] = list.sortWith((a, b) => a._2 > b._2).take(3)
      tuples
    })

    resultRdd.collect().foreach(println)

    // 关闭连接
    sc.stop()

  }

  /**
   * 打印数据，查看原始数据
   */
  def printContent(agentFileRdd: RDD[String]): Unit = {
    val agentArray: Array[String] = agentFileRdd.collect()
    for (i <- agentArray.indices) {
      println("第 " + (i + 1) + "行数据：" + agentArray(i))
    }
  }

}
